package com.idanchuang.count.listener;

import com.alibaba.fastjson.JSON;
import com.google.common.eventbus.EventBus;
import com.idanchuang.count.core.entity.KafkaSendEvent;
import com.idanchuang.count.core.entity.SimpleCountChangeEvent;
import com.idanchuang.count.core.entity.SimpleCountIncrEvent;
import com.idanchuang.count.core.entity.SimpleIncrementCommand;
import com.idanchuang.count.core.model.BizConfig;
import com.idanchuang.count.core.model.BizProperties;
import com.idanchuang.count.dao.CountDao;
import com.idanchuang.count.support.MetricCollector;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.idanchuang.count.core.constant.CountConstants.UID_SET_VALUE;
import static com.idanchuang.count.core.util.JacksonUtil.toJsonStr;
import static com.idanchuang.count.core.util.keyformat.SimpleCountKeysFormatter.formatCountKey;
import static com.idanchuang.count.core.util.keyformat.SimpleCountKeysFormatter.formatUidKey;

/**
 *
 * https://access.feishu.cn/wiki/wikcnZxHnUKnSsPYt48WTvygie6
 *
 * 可以把接入业务分为两类：
 * 1. 对准确性要求高，对实时性有一定的容忍度
 * 计算策略：按照时间窗口聚合，比如5分钟一条
 * 消费策略：kafka单笔消费 + 幂等 + 单次写入
 *
 * topic 命名规则：base_count_single_{bizCode}
 * 如学习中心需求，10分钟聚合一次发送，base_count_single_content_study
 *
 * 2. 对实时性要求高，对准确性有一定的容忍度
 * 计算策略：来一条产生一条消息(实际是1s聚合)
 * 消费策略：kafka批量消费 + bizNo聚合 + 批量写入
 *
 * topic 命名规则：base_count_batch_{bizCode}
 * 如社区pv需求，实时发送，批量消费 base_count_batch_content_community
 *
 * @author Richard_yyf
 * @version 1.0 2021/9/24
 */
@Component
@Slf4j
public class SimpleCountIncrEventListener {

    @Resource
    private BizProperties bizProperties;

    @Resource
    private CountDao countDao;

    @Resource
    private EventBus kafkaSendEventBus;

    /**
     * 幂等键失效时间，秒
     * 默认 1天
     */
    @Value("${count.simple.idempotent-expire:86400}")
    private String countCommandIdempotentExpire;


    /**
     * key [uuidKey, countKey]
     * val [uuidVal, expireSeconds, incrVal]
     * 1. 判断uuid是否存在，如果存在，返回0，代表没有操作过
     * 2. 执行 incrBy 操作
     * 3. 设置uuid，设置过期时间
     * 4. 操作成功，返回incr之后的值
     * local val = redis.call("GET", KEYS[1]);
     * if val == ARGV[1] then return 0;
     * end;
     * local countVal = redis.call('incrby', KEYS[2], ARGV[3]);
     * redis.call('set', KEYS[1], ARGV[1]);
     * redis.call('expire', KEYS[1], tonumber(ARGV[2]));
     * return countVal;
     */

    private static final String LUA_SCRIPT = "local val = redis.call(\"GET\", KEYS[1]);if val == ARGV[1] then return 0;end;local countVal = redis.call('incrby', KEYS[2], ARGV[3]);redis.call('set', KEYS[1], ARGV[1]);redis.call('expire', KEYS[1], tonumber(ARGV[2]));return countVal;";

    /**
     * 判断uuid是否存在，如果存在，返回0，代表没有操作过
     */
    private static final long NOT_HANDLE = 0L;

    private static final DefaultRedisScript<Long> REDIS_SCRIPT = new DefaultRedisScript<>(LUA_SCRIPT, Long.class);

    @KafkaListener(
            topicPattern = "base_count_single.*",
            groupId = "#{'${spring.kafka.consumer.group-id}'}",
            containerFactory = "simpleCountSingleConsumeContainerFactory",
            clientIdPrefix = "simple-count"
    )
    public void singleConsumeIncrEvent(@Payload(required = false) SimpleCountIncrEvent incrEvent,
                                       Acknowledgment ack) {
        // single 对应 非实时，聚合场景
        if (incrEvent == null) {
            ack.acknowledge();
            return;
        }

        log.info("收到singleConsumeIncrEvent: {}", JSON.toJSONString(incrEvent));

        BizConfig bizConfig = bizProperties.getSingleConfigNotThrowEx(incrEvent.getBizCode());
        if (bizConfig == null) {
            log.error("ConsumeIncrEvent 不存在的bizCode: {}", incrEvent.getBizCode());
            ack.acknowledge();
            return;
        }

        String countKey = formatCountKey(bizConfig.getId(), incrEvent.getBizNo());

        Long result;
        String uidKey = formatUidKey(bizConfig.getId(), incrEvent.getBizNo(), incrEvent.getTime());
        try {
            result = handleIdempotentIncr(uidKey, countKey, incrEvent.getIncrement());
        } catch (Exception e) {
            log.error("idempotentIncr 执行失败", e);
            throw e;
        }
        if (result == NOT_HANDLE) {
            // 请求被幂等了
            ack.acknowledge();
            return;
        }

        // 走到这里的result代表自增之后的值
        if (bizConfig.isCountChangeNotifyEnable()) {
            // 是否需要变更通知
            // TODO 服务稳定后删除日志
            log.info("ConsumeIncrEvent成功，开始通知 bizCode[{}]", incrEvent.getBizCode());
            publishCountChangeEvent2NotifyTopic(incrEvent.getBizCode(), incrEvent.getBizNo(),
                    result, bizConfig.getCountChangeNotifyTopic());
        }

        ack.acknowledge();
    }

    @KafkaListener(
            topicPattern = "base_count_batch.*",
            groupId = "#{'${spring.kafka.consumer.group-id}'}",
            containerFactory = "simpleCountBatchConsumeContainerFactory",
            clientIdPrefix = "simple-count-batch"
    )
    public void batchConsumeIncrEvent(@Payload(required = false) List<SimpleCountIncrEvent> incrEvents,
                                                        Acknowledgment ack) {
        // batch 对应 实时，来一条发一条场景
        log.info("收到batchConsumeIncrEvent, {}", JSON.toJSONString(incrEvents));

        // Map<bizCode, Map<bizNo, 聚合incrVal>>
        Map<String, Map<String, Integer>> groupMap = groupIncrEvents(incrEvents);

        List<SimpleIncrementCommand> commands = new ArrayList<>(incrEvents.size());

        // 备注：测试后发现。在监听多个 topic 情况下，
        //  consumer 每次 batch poll 都是同个 topic的一批数据（下一次poll 变成另外一个），
        //  不会出现一批里面有多个topic，所以下面的分组逻辑会有些复杂
        for (Map.Entry<String, Map<String, Integer>> entry : groupMap.entrySet()) {
            BizConfig bizConfig = bizProperties.getSingleConfigNotThrowEx(entry.getKey());
            if (bizConfig == null) {
                log.error("batchConsumeIncrEvent 跳过不存在的bizCode: {}", entry.getKey());
                continue;
            }
            List<SimpleIncrementCommand> commandsForBizCode = entry.getValue()
                    .entrySet()
                    .stream()
                    .map(countEntry -> {
                        // bizCode - entry.getKey(); bizNo - countEntry.getKey()
                        SimpleIncrementCommand incrementCommand = new SimpleIncrementCommand(formatCountKey(bizConfig.getId(), countEntry.getKey()), countEntry.getValue());
                        incrementCommand.setBizCode(entry.getKey());
                        incrementCommand.setBizNo(countEntry.getKey());
                        return incrementCommand;
                    })
                    .collect(Collectors.toList());
            commands.addAll(commandsForBizCode);
        }

        if (commands.size() == 0) {
            ack.acknowledge();
            return;
        }

        // TODO 服务稳定后删除日志
        log.info("收到batchConsumeIncrEvent, bizCode:[{}], size[{}], 实际执行的incrCommandsSize:[{}]", groupMap.keySet(), incrEvents.size(), commands.size());

        log.info("incr commands, {}", JSON.toJSONString(commands));
        List<Long> incrResult;
        try {
            incrResult = countDao.batchIncr(commands);
        } catch (Exception e) {
            log.error("batch incr 执行失败", e);
            throw e;
        }

        for (int i = 0; i < commands.size(); i++) {
            Long result = incrResult.get(i);
            if (result == null) {
                log.error("batchIncr result return null, unexpected, command info: {}", commands.get(i));
                continue;
            }
            BizConfig bizConfig = bizProperties.getSingleConfigNotThrowEx(commands.get(i).getBizCode());
            if (!bizConfig.isCountChangeNotifyEnable()) {
                continue;
            }
            publishCountChangeEvent2NotifyTopic(commands.get(i).getBizCode(), commands.get(i).getBizNo(), result, bizConfig.getCountChangeNotifyTopic());
        }

        ack.acknowledge();
    }

    private Map<String, Map<String, Integer>> groupIncrEvents(List<SimpleCountIncrEvent> incrEvents) {
        Map<String, Map<String, Integer>> groupMap = new HashMap<>(8);

        for (SimpleCountIncrEvent incrEvent : incrEvents) {
            if (incrEvent == null) {
                continue;
            }
            Map<String, Integer> countMap = groupMap.computeIfAbsent(incrEvent.getBizCode(), key -> new HashMap<>(incrEvents.size()));
            Integer val = incrEvent.getIncrement() + countMap.getOrDefault(incrEvent.getBizNo(), 0);
            countMap.put(incrEvent.getBizNo(), val);
        }

        return groupMap;
    }

    private Long handleIdempotentIncr(String uidKey, String countKey, Integer increment) {
        // key [uuidKey, countKey]
        // val [uuidVal, expireSeconds, incrVal]
        return countDao.executeScript(REDIS_SCRIPT, Arrays.asList(uidKey, countKey),
                UID_SET_VALUE, countCommandIdempotentExpire, String.valueOf(increment));
    }

    private void publishCountChangeEvent2NotifyTopic(String bizCode, String bizNo,
                                                     Long incrResult, String topic) {

        // TODO 服务稳定后删除日志
        log.info("publishCountChangeEvent, bizCode: {}, bizNo: {}, val: {}", bizCode, bizNo, incrResult);

        SimpleCountChangeEvent event = new SimpleCountChangeEvent(bizCode, Math.toIntExact(incrResult), bizNo);
        // 根据 bizNo 分区
        ProducerRecord<Object, Object> record = new ProducerRecord<>(topic, bizNo, toJsonStr(event));
        // 不保证百分百的发送成功率
        kafkaSendEventBus.post(new KafkaSendEvent(record, new ListenableFutureCallback<SendResult<Object, Object>>() {
            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                MetricCollector.collectAfterEventSent("simple_count_change");
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("发送消息至kafka失败, command: " + event.toString(), ex);
                MetricCollector.collectAfterEventSentFail("simple_count_change");
            }
        }));
    }

}
